package log

import (
	"context"
	"errors"
	"gitee.com/kinwyb/appTools/common/exit"
	"github.com/olivere/elastic/v7"
	"github.com/sirupsen/logrus"
	"gopkg.in/sohlich/elogrus.v7"
	"net"
	"strings"
	"time"
)

var esClient *elastic.Client
var esHook *elasticHook
var logEsHost string

func init() {
	exit.Listen(func(args ...interface{}) {
		if esClient != nil {
			esClient.Stop()
		}
		if esHook != nil {
			esHook.Cancel()
		}
	})
}

// 获取一个本地IP地址
func localIPv4() string {
	var ips []string
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return ""
	}
	var ip string
	for _, a := range addrs {
		if ipnet, ok := a.(*net.IPNet); ok && !ipnet.IP.IsLoopback() && ipnet.IP.To4() != nil {
			networkSize, _ := ipnet.Mask.Size()
			if ipnet.Mask.String() == "255.255.255.0" ||
				ipnet.Mask.String() == "ffffff00" ||
				networkSize == 24 {
				ip = ipnet.IP.String()
				break
			}
			ips = append(ips, ipnet.IP.String())
		}
	}
	if ip != "" {
		return ip
	} else if len(ips) > 0 {
		return ips[0]
	}
	return ""
}

// ToElasticsearch 输出到elasticsearch
// indexWithDay 按天分割elastic index
func ToElasticsearch(logName string, level logrus.Level, indexWithDay ...bool) {
	logName = strings.ToLower(logName)
	logEsHost = strings.TrimSpace(logEsHost)
	if logEsHost == "" {
		return
	}
	host := localIPv4()
	var err error
	esClient, err = elastic.NewClient(
		elastic.SetURL(logEsHost),
		elastic.SetSniff(false),
		elastic.SetErrorLog(logrus.New()),
	)

	if err != nil {
		AppTool.Errorf("elasticsearch客户端创建失败:%s", err.Error())
		return
	}
	// 可以按日期
	esHook, err = newBulkProcessorElasticHookWithFunc(esClient, host, level, func() string {
		if len(indexWithDay) > 0 && indexWithDay[0] {
			return logName + "_" + time.Now().Format("2006-01-02")
		} else {
			return logName
		}
	})
	if err != nil {
		AppTool.Errorf("elasticsearch日志输出失败:%s", err.Error())
	} else {
		//lg.SetReportCaller(true)
		lg.Hooks.Add(esHook)
	}
}

// 源码地址: https://github.com/sohlich/elogrus/tree/v7
// 由于源码中批量同步不能修改成按时间,所以做了一些调整

// elasticHook is a logrus hook for ElasticSearch
type elasticHook struct {
	client       *elastic.Client
	host         string
	index        elasticIndexNameFunc
	levels       []logrus.Level
	ctx          context.Context
	ctxCancel    context.CancelFunc
	fireFunc     fireFunc
	msgChannel   chan *logrus.Entry
	timeOut      *time.Ticker
	errCount     int64         //错误发送次数
	maxErrCount  int64         //最大错误次数，达到最大错误次数后会进入错误状态,错误状态下数据不往elastic发送
	errState     bool          //错误状态
	errStateTime time.Duration //错误状态持续时间，超过错误状态时间后调整回正常
}

// elasticIndexNameFunc get index name for elastic
type elasticIndexNameFunc func() string

type fireFunc func(entry *logrus.Entry, hook *elasticHook) error

type message struct {
	Host      string
	Timestamp string `json:"@timestamp"`
	Message   string
	Data      logrus.Fields
	Level     string
}

// newBulkProcessorElasticHookWithFunc creates new hook with
// function that provides the index name. This is useful if the index name is
// somehow dynamic especially based on time that uses a bulk processor for
// indexing.
// client - ElasticSearch client with specific es version (v5/v6/v7/...)
// host - host of system
// level - log level
// indexFunc - function providing the name of index
func newBulkProcessorElasticHookWithFunc(client *elastic.Client, host string, level logrus.Level, indexFunc elasticIndexNameFunc) (*elasticHook, error) {
	fireFunc, err := makeBulkFireFunc(client)
	if err != nil {
		return nil, err
	}
	return newHookFuncAndFireFunc(client, host, level, indexFunc, fireFunc)
}

func newHookFuncAndFireFunc(client *elastic.Client, host string, level logrus.Level, indexFunc elasticIndexNameFunc, fireFunc fireFunc) (*elasticHook, error) {
	var levels []logrus.Level
	for _, l := range []logrus.Level{
		logrus.PanicLevel,
		logrus.FatalLevel,
		logrus.ErrorLevel,
		logrus.WarnLevel,
		logrus.InfoLevel,
		logrus.DebugLevel,
	} {
		if l <= level {
			levels = append(levels, l)
		}
	}

	ctx, cancel := context.WithCancel(context.TODO())

	// Use the IndexExists service to check if a specified index exists.
	exists, err := client.IndexExists(indexFunc()).Do(ctx)
	if err != nil {
		// Handle error
		cancel()
		return nil, err
	}
	if !exists {
		createIndex, err := client.CreateIndex(indexFunc()).Do(ctx)
		if err != nil {
			cancel()
			return nil, err
		}
		if !createIndex.Acknowledged {
			cancel()
			return nil, elogrus.ErrCannotCreateIndex
		}
	}

	ret := &elasticHook{
		client:       client,
		host:         host,
		index:        indexFunc,
		levels:       levels,
		ctx:          ctx,
		ctxCancel:    cancel,
		fireFunc:     fireFunc,
		msgChannel:   make(chan *logrus.Entry, 100),
		timeOut:      time.NewTicker(5 * time.Second),
		maxErrCount:  10,
		errState:     false,
		errStateTime: 2 * time.Minute,
	}
	go ret.sendMsgToElastic()
	return ret, nil
}

func createMessage(entry *logrus.Entry, hook *elasticHook) *message {
	level := entry.Level.String()
	data := make(logrus.Fields)
	for k, v := range entry.Data {
		if k == logrus.ErrorKey {
			if err, ok := v.(error); ok {
				data[k] = err.Error()
			}
		} else {
			data[k] = v
		}
	}
	return &message{
		hook.host,
		entry.Time.UTC().Format(time.RFC3339Nano),
		entry.Message,
		data,
		strings.ToUpper(level),
	}
}

// Create closure with bulk processor tied to fireFunc.
func makeBulkFireFunc(client *elastic.Client) (fireFunc, error) {
	processor, err := client.BulkProcessor().
		Name("elogrus.v3.bulk.processor").
		Workers(2).
		BulkSize(-1).
		BulkActions(-1).
		Backoff(elastic.NewSimpleBackoff(5)). //重试次数5次
		FlushInterval(5 * time.Second).
		Do(context.Background())

	return func(entry *logrus.Entry, hook *elasticHook) error {
		r := elastic.NewBulkIndexRequest().
			Index(hook.index()).
			//Type("log").
			Doc(*createMessage(entry, hook))
		processor.Add(r)
		return nil
	}, err
}

func (hook *elasticHook) sendMsgToElastic() {
	if hook.msgChannel == nil {
		hook.msgChannel = make(chan *logrus.Entry, 10)
	}
	for {
		select {
		case msg := <-hook.msgChannel:
			hook.fireFunc(msg, hook)
		case <-hook.ctx.Done():
			return
		}
	}
}

// Fire is required to implement
// Logrus hook
func (hook *elasticHook) Fire(entry *logrus.Entry) error {
	if hook.errState {
		return errors.New("elastic日志异常")
	}
	select {
	case hook.msgChannel <- entry:
		hook.errCount = 0
		return nil
	default:
		select {
		case <-hook.timeOut.C:
			hook.errCount++
			if hook.errCount > hook.maxErrCount {
				hook.errState = true
				go func(ts time.Duration) {
					t := time.NewTimer(ts)
					<-t.C
					hook.errState = false
					hook.errCount = 0
				}(hook.errStateTime)
			}
			return errors.New("elastic日志异常: 数据发送超时")
		case <-hook.ctx.Done():
			return nil
		}
	}
}

// Levels Required for logrus hook implementation
func (hook *elasticHook) Levels() []logrus.Level {
	return hook.levels
}

// Cancel all calls to elastic
func (hook *elasticHook) Cancel() {
	hook.ctxCancel()
}
